package com.bsj.travel.mqtt.listener;

import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import com.bsj.travel.config.rocketmq.producer.BaseMqPushUtil;
import com.bsj.travel.constant.BaseServerTopicConstants;
import com.bsj.travel.def.common.DTO.mqtt.EventAct;
import com.bsj.travel.def.common.DTO.mqtt.MqttEvent;
import com.bsj.travel.def.common.DTO.mqtt.NetworkEnum;
import lombok.extern.slf4j.Slf4j;
import net.dreamlu.iot.mqtt.codec.MqttPublishMessage;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.tio.core.ChannelContext;

import java.nio.charset.StandardCharsets;

/**
 * @author yinhao
 * @version 1.0
 * @description 客户端消息监听
 * @date 2023/11/7
 */
@Slf4j
public class MqttServerMessageListener implements IMqttMessageListener {

    @Autowired
    private BaseMqPushUtil baseMqPushUtil;

    @Override
    public void onMessage(ChannelContext context, String clientId, String topic, MqttQoS qoS, MqttPublishMessage message) {
        String msg = new String(message.getPayload(), StandardCharsets.UTF_8);
        log.info("clientId: {} topic: {} payload: {}", clientId, topic, msg);
        long millis = System.currentTimeMillis();
        MqttEvent<String> mqttEvent = new MqttEvent<>(millis, clientId, topic, qoS.value(), msg);
        EventAct eventAct = EventAct.builder()
                .timestamp(millis)
                .traceId(RandomUtil.randomString(30))
                .serveType(0)
                .operate(topic)
                .event(mqttEvent)
                .code(NetworkEnum.SUCCESS.getCode())
                .build();
        //消息转发给MQ，按业务处理
        baseMqPushUtil.sendTopicBytes(BaseServerTopicConstants.TRAVEL_MQTT_MSG_PROCESS_TOPIC,
                BaseServerTopicConstants.TRAVEL_MQTT_MSG_PROCESS_TAG, JSON.toJSONBytes(eventAct));
    }
}
